package edu.ncsu.csc.streamingmr;


import java.io.ByteArrayOutputStream;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.net.Socket;

public class Client extends Thread{
	int tid;
	/*private String hostname;
	private int port;*/
	private FileInputStream fin;
	private ByteArrayOutputStream buffer=new ByteArrayOutputStream();
	private byte[] content;
	private OutputStream out;
	private int iters = 0;
	private int sleepTime = 0;


	public Client(int id, String hostname, int port, String filename, int iters, int sleepTime) {
		this.tid=id;
		this.iters = iters; 
		this.sleepTime = sleepTime;
		/*this.hostname=hostname;
		this.port=port;*/
		
		Socket MyClient;
	    try {
	           MyClient = new Socket(hostname, port);
	           this.out=MyClient.getOutputStream();
	           fin=new FileInputStream(filename);
	           byte[] bytes=new byte[32];
	           int r=fin.read(bytes);
	           while(r!=-1) {
	        	   buffer.write(bytes, 0, r);
	        	   r=fin.read(bytes);
	           }
	           content=buffer.toByteArray();
			   System.out.println(new String(content));
	    }
	    catch (IOException e) {
	        System.out.println(e);
	    }

	}
	public void run() {

		try {
			//this.out.write("!BEGIN!\n".getBytes());
			for(int i=0;i<iters;i++) {
				this.out.write(content);
				this.out.flush();
				if(sleepTime>0)
					Thread.sleep(sleepTime);
				System.out.println("sent "+i);
			}
			//this.out.write("!END!\n".getBytes());
			System.out.println(new String(content));
		}catch(Exception e) {
			e.printStackTrace();
		}finally {
			try{
				out.close();
			}catch(Exception e) {
				e.printStackTrace();
			}
		}
		
	}
	
	private static int numberthreads;
	
	/**
	 * @param args
	 */
	public static void main(String[] args) {
		numberthreads=new Integer(args[0]).intValue();
		String hostname;
		int port;
		hostname=args[1];
		port=new Integer(args[2]).intValue();
		int iters = new Integer(args[4]).intValue();
		String fname=args[3];
<<<<<<< .mine
		//boolean start=false;
=======
		
>>>>>>> .r211
		Thread[] threads=new Thread[numberthreads];
		for(int i=0;i<numberthreads;i++) {
<<<<<<< .mine
			threads[i]=new Client(i, hostname, port, false, fname, iters, Integer.parseInt(args[5]));
=======
			threads[i]=new Client(i, hostname, port, fname, iters, Integer.parseInt(args[5]));
>>>>>>> .r211
			threads[i].start();
		}		
		for(int i=0;i<numberthreads;i++) {
			try {
				threads[i].join();
			}catch(InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

}
